package com.thghh.executor;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 包名称：com.thghh.executor
 * 类名称：MeantimeCompletionService
 * 类描述：一组任务在某个时间点同时完成的服务提供者,允许一个或多个线程等待，直到在其它线程中执行的一组操作完成。
 * 创建人：@author pengzhikang
 * 创建时间：2021/9/3 9:31
 */
@Slf4j
public class MeantimeCompletionService {

    /** 自动生成group id计数 */
    private static AtomicInteger groupIdCount = new AtomicInteger();

    /** 当前组内任务数量 */
    private int taskSize;
    /** 当前组id */
    private int groupId;
    /** 组内任务开始执行时间 */
    private long startTime;

    /** 当前的任务数量 */
    private AtomicInteger currentTaskSize;
    /** 任务池调度器 */
    private ThreadPoolExecutor taskPoolScheduler;
    /** 倒计时门闩 */
    private CountDownLatch countDownLatch;

    public MeantimeCompletionService(int taskSize, ThreadPoolExecutor taskPoolScheduler) {
        this.groupId = getNextGroupId();
        this.taskSize = taskSize;
        this.taskPoolScheduler = taskPoolScheduler;
        this.countDownLatch = new CountDownLatch(taskSize);
        this.currentTaskSize = new AtomicInteger();
    }

    /**
     * 获取group id
     * @return long group id
     * @author pengzhikang
     * @date 2021/9/3 10:40
     */
    private int getNextGroupId() {
        return groupIdCount.incrementAndGet();
    }

    /**
     * 提交任务
     *
     * @param task 任务
     * @return Future<T> 返回值
     * @author pengzhikang
     * @date 2021/9/3 10:43
     */
    public <T> Future<T> submit(Task<T> task) {
        int size = currentTaskSize.incrementAndGet();
        if (size > taskSize) {
            throw new IllegalStateException("运行" + size + "个任务大于任务最大容量" + taskSize);
        }
        if (startTime == 0L) {
            startTime = System.currentTimeMillis();
        }
        TaskAdapter<T> taskAdapter = new TaskAdapter<>(groupId, size, countDownLatch, task);
        return taskPoolScheduler.submit(taskAdapter);
    }


    /**
     * 等待组内任务完成
     *
     * @throws InterruptedException 线程中断异常
     * @author pengzhikang
     * @date 2021/9/3 11:51
     */
    public void await() throws InterruptedException {
        int size = currentTaskSize.get();
        if (size != taskSize) {
            throw new IllegalStateException("运行的" + size + "个任务和指定任务容量" + taskSize + "不匹配");
        }
        log.info("任务组{}等待{}个任务完成", groupId, size);
        countDownLatch.await();
        long time = System.currentTimeMillis() - startTime;
        log.info("任务组{}共{}个任务完成总耗时{}ms", groupId, size, time);
    }

}